#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include "logger.hpp"


// RabbitMQ客户端封装类
// 提供AMQP协议操作功能，支持消息发布和订阅
// 使用libev事件循环实现异步IO处理
class MQclient
{
private:
    struct ev_loop *_loop; // 事件监控
    std::unique_ptr<AMQP::LibEvHandler> _handler;  // AMQP库的事件处理适配器
    std::unique_ptr<AMQP::TcpConnection> _conn;    // AMQP 连接
    std::unique_ptr<AMQP::TcpChannel> _channel;    // AMQP通信通道
    struct ev_async _async_watcher;                // 异步事件监视器，用于线程间通信
    std::thread _loop_thread;                      // 运行事件循环的工作线程

public:
    // 消息处理回调函数类型定义，用于处理接收到的消息
    using MessageCallback = std::function<void(const char *, size_t)>;

    using ptr = std::shared_ptr<MQclient>;

    MQclient(const std::string &user,   // user 用户名
             const std::string &pw,     // pw 密码
             const std::string &host)   // host RabbitMQ服务器地址和端口，格式: "host:port"
    {
        // 直接用 EV_DEFAULT 宏初始化事件监控句柄
        _loop = EV_DEFAULT;
        // 创建AMQP事件处理器，将AMQP库与监控句柄关联
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);

        //构建AMQP连接URL address参数格式 "amqp://root:123456@127.0.0.1:5672/"
        std::string url = "amqp://" + user + ":" + pw + "@" + host + "/";
        AMQP::Address address(url);

        // 创建AMQP TCP连接
        _conn = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);

        // 创建AMQP通道
        _channel = std::make_unique<AMQP::TcpChannel>(_conn.get());

        // 初始化异步事件监视器，用于通知事件循环停止
        ev_async_init(&_async_watcher, watcher_callback);
        ev_async_start(_loop, &_async_watcher);

        // 启动事件循环线程，负责处理所有AMQP相关的异步事件
        _loop_thread = std::thread([this]()
                                   { ev_run(_loop, 0); });
    }

    ~MQclient()
    {
        // 通知事件循环停止
        ev_async_send(_loop, &_async_watcher);
        if (_loop_thread.joinable())
        {
            _loop_thread.join();
        }
        // 清理资源
        ev_async_stop(_loop, &_async_watcher);
        _loop = nullptr;
    }
    // 配置交换器、队列和绑定关系  
    // ex_name - 交换机名称   que_name - 队列名称   routing_key - 绑定名称 默认为"routing_key"   ex_type- -交换器类型
    void compoents(const std::string &ex_name,
                   const std::string &que_name,              
                   const std::string routing_key = "routing_key",
                   AMQP::ExchangeType ex_type = AMQP::ExchangeType::direct)
    {
        // 创建交换机
        AMQP::Deferred &exchange_defferred = _channel->declareExchange(ex_name, ex_type);
        // Deferred类型需要设置回调函数，成功、错误、完成后...
        exchange_defferred.onError([](const char *msg)
                                   {
                                        LOG_ERROR("申明交换机失败 : {}", msg);
                                        exit(1); });
        
        exchange_defferred.onSuccess([ex_name]()
                                     { LOG_INFO("申明交换机 {} 成功!", ex_name); });

        // 创建队列
        AMQP::Deferred &queue_defferred = _channel->declareQueue(que_name);
        queue_defferred.onError([](const char *msg)
                                {
                                    LOG_ERROR("申明队列失败,原因: {}", msg);
                                    exit(1); });

        queue_defferred.onSuccess([que_name]()
                                  { LOG_INFO("申明队列 {} 成功!", que_name); });

        // 绑定交换机和队列
        AMQP::Deferred &bind_defferred = _channel->bindQueue(ex_name, que_name, routing_key);
        bind_defferred.onError([ex_name, que_name](const char *msg)
                               {
                                    LOG_ERROR("绑定队列 {}--{} 失败,原因: {}", ex_name, que_name, msg);
                                    exit(1); });

        bind_defferred.onSuccess([ex_name, que_name]()
                                 { LOG_INFO("绑定队列 {}--{} 成功!", ex_name, que_name); });
    }

    // 消息发布
    bool publish(const std::string &exchange,
                 const std::string &msg,
                 const std::string &routing_key = "routing_key")
    {
        bool ret = _channel->publish(exchange, routing_key, msg.c_str());
        if (!ret)
        {
            LOG_ERROR("{} 交换机发布消息失败", exchange);
            return false;
        }
        return true;
    }

    // 订阅队列消息 queue 要订阅的队列名称   cb 消息处理回调函数
    void consumer(const std::string &queue, const MessageCallback &cb)
    {
        LOG_INFO("开始订阅 {} 队列消息\n", queue);
        _channel->consume(queue, AMQP::noack) // 使用 AMQP::noack 或手动确认
            .onReceived([this, cb](const AMQP::Message &message,
                                   uint64_t deliveryTag,
                                   bool redelivered)
                        {
                            cb(message.body(), message.bodySize());
                            // 如果未使用 AMQP::noack，需要手动确认
                            _channel->ack(deliveryTag); })
            .onError([queue](const char *message)
                     {
                        LOG_ERROR("订阅 {} 队列消息失败: {}\n", queue, message);
                        exit(0); });
    }

private:
    // 异步事件监视器回调函数
    // 当接收到异步事件时，终止事件循环
    // loop 事件循环句柄   watcher 异步监视器指针  revents 事件标志
    static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents)
    {
        ev_break(loop, EVBREAK_ALL);
    }
};